package com.alison.datastream.chapter3_sink;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @Author alison
 * @Date 2024/4/8 23:09
 * @Version 1.0
 * @Description
 */
public class S2_SinkTest2_Redis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String inputPath = "D:/workspace/lab/learnbigdata/learnflink/flink-datastream/src/main/resources/datastream/sensor.txt";
        // 从文件读取数据
        DataStream<String> inputStream = env.readTextFile(inputPath);
        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        /*
        FlinkJedisPoolConfig类：Jedis连接池的相关参数；
        RedisMapper接口：从用户数据中提取键值，并构成Redis命令的映射器，需要用户自己实现；
        RedisSink类：根据构建好的FlinkJedisPoolConfig和RedisMapper将流数据写入Redis。
         */
        // 定义jedis连接配置(我这里连接的是docker的redis)
        FlinkJedisPoolConfig config = new FlinkJedisPoolConfig.Builder()
                .setHost("192.168.56.101")
                .setPort(6379)
                .setPassword("redis.com")
                .setDatabase(0)
                .setTimeout(60000)
                .build();
        dataStream.addSink(new RedisSink<>(config, new MyRedisMapper()));
        env.execute();
    }
    // 自定义RedisMapper
    public static class MyRedisMapper implements RedisMapper<SensorReading> {
        // 定义保存数据到redis的命令，存成Hash表，hset sensor_temp id temperature
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "sensor");
        }

        @Override
        public String getKeyFromData(SensorReading data) {
            return data.getId();
        }

        @Override
        public String getValueFromData(SensorReading data) {
            return String.valueOf(data.getTemperature());
        }
    }
}
